package com.bestcem.xm.txtanls.job;

import com.bestcem.xm.common.core.constant.Constants;
import com.bestcem.xm.common.core.domain.web.ServiceResult;
import com.bestcem.xm.componet.redis.service.RedisService;
import com.bestcem.xm.componet.xxljob.annotation.XxlRegister;
import com.bestcem.xm.txtanls.dao.AnalysisProjectDao;
import com.bestcem.xm.txtanls.entity.pojo.ProjectDO;
import com.bestcem.xm.txtanls.enums.AnalysisStatusEnum;
import com.bestcem.xm.txtanls.enums.TrainStatusEnum;
import com.bestcem.xm.txtanls.service.AsyncTaskService;
import com.bestcem.xm.txtanls.util.AnalysisUtil;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.redisson.api.RLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

@Component
public class TxtanlsJobService {

    private static final Logger logger = LoggerFactory.getLogger(TxtanlsJobService.class);

    private static final String TRAIN_JOB_LOCK_PREFIX = "txtanls:train";

    @Resource
    private AnalysisProjectDao analysisProjectDao;
    @Resource
    private AsyncTaskService asyncTaskService;
    @Resource
    private AnalysisUtil analysisUtil;
    @Resource
    private RedisService redisService;

    @XxlJob(value = "txtanls-created-job")
    //@XxlRegister(cron = "*/5 * * * * ?", triggerStatus = 1, jobDesc = "实时训练定时任务")
    @XxlRegister(cron = "${app.txtanls.task.created-cron:*/5 * * * * ?}", triggerStatus = 1, jobDesc = "实时训练定时任务")
    public void createdTrainingJob() {
        if (logger.isInfoEnabled()) {
            logger.info("[Txtanls] 实时训练定时任务开始");
        }
        String jobName = "txtanls:created";
        String projectId;

        ProjectDO project = analysisProjectDao.findOneByStatus(AnalysisStatusEnum.CREATED.getIndex());
        if (Objects.isNull(project)) {
            if (logger.isInfoEnabled()) {
                logger.info("[Txtanls] 实时训练定时任务无满足目标");
            }
            return;
        }

        projectId = project.getId();
        // 获取锁
        RLock lock = this.redisService.getLock(this.getTrainingJobLockKey(projectId));
        if (lock.isLocked()) {
            if (logger.isInfoEnabled()) {
                logger.info("[Txtanls] {} 项目已在实时训练", projectId);
            }
            return;
        }

        // 加锁成功与否
        boolean isLocked = false;
        try {
            // 加锁, 不自动释放
            if (lock.tryLock(0L, TimeUnit.SECONDS)) {
                isLocked = true;
                // 训练
                ServiceResult<Object> serviceResult = asyncTaskService.trainTask(project);
                if (!serviceResult.isSuccess()) {
                    ErrorHandler(projectId, jobName, serviceResult.getMsg(), null);
                }
            } else { // 加锁失败
                if (logger.isInfoEnabled()) {
                    logger.info("[Txtanls] {} 项目已在实时训练", projectId);
                }
            }
        } catch (InterruptedException e) {
            logger.warn("[Txtanls] {} 项目实时训练获取锁被中断", projectId);
        } catch (Exception e) {
            if (!ObjectUtils.isEmpty(projectId)) {
                ErrorHandler(projectId, jobName, e.getMessage(), e);
            }
        } finally {
            if (isLocked) {
                // 是否当前线程持有锁
                if (lock.isHeldByCurrentThread()) {
                    // 手动释放锁
                    lock.unlock();
                } else {
                    logger.error("[Txtanls] {} 项目实时训练锁持有对象转移", projectId);
                }
            }
        }

        if (logger.isInfoEnabled()) {
            logger.info("[Txtanls] 实时训练定时任务完成; ProjectId: {}", projectId);
        }
    }

    @XxlJob(value = "txtanls-inactive-job")
    //@XxlRegister(cron = "0 0 0 * * ?", triggerStatus = 1, jobDesc = "增量训练定时任务")
    @XxlRegister(cron = "${app.txtanls.task.inactive-cron:0 0 0 * * ?}", triggerStatus = 1, jobDesc = "增量训练定时任务")
    public void inactiveTrainingJob() {
        if (logger.isInfoEnabled()) {
            logger.info("[Txtanls] 增量训练定时任务开始");
        }
        String jobName = "txtanls:inactive";

        // 查询所有的增量训练的项目
        List<ProjectDO> projectList = analysisProjectDao.findAllByStatus(AnalysisStatusEnum.INACTIVE.getIndex());
        if (CollectionUtils.isEmpty(projectList)) {
            if (logger.isInfoEnabled()) {
                logger.info("[Txtanls] 增量训练定时任务无满足目标");
            }
            return;
        }
        if (logger.isInfoEnabled()) {
            logger.info("[Txtanls] 增量训练定时任务共 {} 条数据", projectList.size());
        }

        for (ProjectDO project : projectList) {
            String projectId = project.getId();
            // 获取锁
            RLock lock = redisService.getLock(this.getTrainingJobLockKey(projectId));
            if (lock.isLocked()) {
                if (logger.isInfoEnabled()) {
                    logger.info("[Txtanls] {} 项目已在增量训练", projectId);
                }
                return;
            }

            boolean isLocked = false;
            try {
                // 加锁, 不自动释放
                if (lock.tryLock(0L, TimeUnit.SECONDS)) {
                    isLocked = true;
                    // 训练
                    ServiceResult<Object> serviceResult = asyncTaskService.trainTask(project);
                    if (!serviceResult.isSuccess()) {
                        ErrorHandler2(projectId, jobName, serviceResult.getMsg(), null);
                    }
                } else { // 加锁失败
                    if (logger.isInfoEnabled()) {
                        logger.info("[Txtanls] {} 项目已在增量训练", projectId);
                    }
                }
            } catch (InterruptedException e) {
                logger.warn("[Txtanls] {} 项目增量训练获取锁被中断", projectId);
            } catch (Exception e) {
                if (!ObjectUtils.isEmpty(projectId)) {
                    ErrorHandler2(projectId, jobName, e.getMessage(), e);
                }
            } finally {
                if (isLocked) {
                    // 是否当前线程持有锁
                    if (lock.isHeldByCurrentThread()) {
                        // 手动释放锁
                        lock.unlock();
                    } else {
                        logger.error("[Txtanls] {} 项目增量训练锁持有对象转移", projectId);
                    }
                }
            }
        }

        if (logger.isInfoEnabled()) {
            logger.info("[Txtanls] 增量训练定时任务全部完成;");
        }
    }


    private void ErrorHandler2(String projectId, String jobName, String errorMsg, Exception e) {
        // 更新项目状态
        int failedUpdateResult = analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.INACTIVE);
        if (failedUpdateResult == 0) {
            logger.error("[Txtanls] {} 失败任务更新项目状态失败", projectId);
        }
        // 更新训练状态
        int updateStateEndResult = analysisProjectDao.updateAnalysisProjectProgress(projectId, TrainStatusEnum.FINISHED, Constants.PROGRESS_FINISHED);
        if (updateStateEndResult == 0) {
            logger.error("[Txtanls] {} 失败任务更新训练状态失败", projectId);
        }
        if (Objects.isNull(e)) {
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg);
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg);
        } else {
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg, e);
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg, e);
        }
    }

    /**
     * 训练任务失败处理
     *
     * @param projectId 项目id
     * @param jobName   任务名
     * @param errorMsg  错误信息
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/7/5 11:17
     */
    private void ErrorHandler(String projectId, String jobName, String errorMsg, Exception e) {
        // 更新项目状态
        int failedUpdateResult = analysisUtil.updateProjectStatus(projectId, AnalysisStatusEnum.INACTIVE);
        if (failedUpdateResult == 0) {
            logger.error("[Txtanls] {} 失败任务更新项目状态失败", projectId);
        }
        // 更新训练状态
        int updateStateEndResult = analysisProjectDao.updateAnalysisProjectProgress(projectId, TrainStatusEnum.FINISHED, Constants.PROGRESS_FINISHED);
        if (updateStateEndResult == 0) {
            logger.error("[Txtanls] {} 失败任务更新训练状态失败", projectId);
        }
        if (Objects.isNull(e)) {
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg);
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg);
        } else {
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg, e);
            logger.error("[Txtanls] {}-{} 增量训练失败; ErrorMsg: {}", jobName, projectId, errorMsg, e);
        }
    }

    /**
     * 获取异步训练任务的锁的key
     *
     * @param projectId 项目id
     * @return Lock key
     * @author XiaoYunSong <ys.xiao@idiaoyan.cn>
     * @date 2021/6/30 14:53
     */
    public String getTrainingJobLockKey(String projectId) {
        return String.format("%s:%s", TRAIN_JOB_LOCK_PREFIX, projectId);
    }
}
